package com.atguigu.chapter07;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/2 10:01
 */
public class Flink32_Checkpoint_Config {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        // ck的清理：默认只保存最近成功的一次checkpoint，可以通过一下参数指定保存的个数：
        // state.checkpoints.num-retained: 1



        // 每 1000ms 开始一次 checkpoint
        env.enableCheckpointing(1000);
//        RocksDBStateBackend rocksDBStateBackend = new RocksDBStateBackend("hdfs://hadoop1:8020/flinktest/checkpoint/");
        FsStateBackend fsStateBackend = new FsStateBackend("hdfs://hadoop1:8020/flinktest/checkpoint");
        env.setStateBackend(fsStateBackend);

        // 高级选项：
        // 设置模式为精确一次 (这是默认值)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 确认 checkpoints 之间的时间会进行 500 ms
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // 同一时间只允许一个 checkpoint 进行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        // 开启在 job 中止后仍然保留的 externalized checkpoints
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);


        /**
         * 任务恢复 重启策略
         */
        // 固定延迟重启策略：重启3次，每次失败后等待10000毫秒
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L));
        // 故障率重启策略：在5分钟内，只能重启5次，每次失败后最少需要等待10秒
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.of(5, TimeUnit.MINUTES), Time.of(10, TimeUnit.SECONDS)));



        env
                .socketTextStream("hadoop1", 9999)
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(word);
                        }
                    }
                })
                .map(value -> Tuple2.of(value, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(0)
                .sum(1)
                .print();

        env.execute();

    }


}


/*
1)	第一条数据来了之后，开启一个 kafka 的事务（transaction），正常写入 kafka 分区日志但标记为未提交，这就是“预提交”
2)	jobmanager 触发 checkpoint 操作，barrier 从 source 开始向下传递，遇到 barrier 的算子将状态存入状态后端，并通知 jobmanagerr
3)	sink 连接器收到 barrier，保存当前状态，存入 checkpoint，通知 jobmanager，并开启下一阶段的事务，用于提交下个检查点的数据
4)	jobmanager 收到所有任务的通知，发出确认信息，表示 checkpoint 完成
5)	sink 任务收到 jobmanager 的确认信息，正式提交这段时间的数据
6)	外部kafka关闭事务，提交的数据可以正常消费了


checkpoint原理： ---- 算法名称： Chandy-Lamport算法   异步分界线快照算法  昌迪兰波特算法
    JobManager内部的一个协调器 周期性的生成 barrier，插入到source前（也就是插入到数据流里）， 随着流程序，向下游广播传递，
    当barrier到达某一个 task 时，会触发当前task的备份，如果上游是多个task，那么会等待所有的task都把barrier传递过来，才会触发（展开：barrier对齐、不对齐）。
    每一个task完成备份的时候，会向 jobmanager里的协调器 发送一个报告消息，告诉他 自己 备份的地址和相关元数据
    当 jobmanager收到 sink的完成通知时，表示本次全局的 checkpoint完成，备份自己的元数据，它会向所有的task发出 checkpoint完成的通知
    （结合 2pc展开讲： 一开始 sink自己 备份的时候，会 预提交。 收到 jobmanager 的完成通知，会进行正式提交。）

 */
